/**
 * 全窗口函数
 *
 *
 *
 *
 *
 */
package com.atguigu.stream.window

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object AvgTempByProcessWindowFunction {


  case class AvgInfo (id :String,avgTemp :Double,windowStart:Long,winowEnd:Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.addSource(new SensorSource)
      .keyBy(_.id)
      .timeWindow(Time.seconds(10))
      .process(new AvgTempFunc)
      .print()

    env.execute()

  }


  //第一个输入类型，输出类型，key的值 ，聚合那个窗口
  class AvgTempFunc extends ProcessWindowFunction[SensorReading,AvgInfo,String,TimeWindow]{
    //窗口闭合时调用
    override def process(key: String, context: Context, elements: Iterable[SensorReading], out: Collector[AvgInfo]): Unit = {
      val count = elements.size
      var sum = 0.0
      for (r <- elements){
        sum += r.temperature
      }

      //单位mm
      var windowStart = context.window.getStart
      var windowEnd =context.window.getEnd


      out.collect(AvgInfo(key,sum/count,windowStart,windowEnd))






    }
  }

}
